package com.neusoft.databus.agent.pool.socket;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;

import org.apache.commons.pool2.PooledObject;
import org.springframework.stereotype.Component;

import com.neusoft.databus.common.connection.SocketConnection;
import com.neusoft.databus.common.pool.AbstractPoolFactory;

import lombok.extern.slf4j.Slf4j;

/**
 * @author suncf
 */
@Slf4j
@Component

public class SocketFactory extends AbstractPoolFactory<SocketConnection, Socket> {
    @Override
    public Socket create(final SocketConnection connection) throws IOException {
        if (SocketFactory.log.isInfoEnabled()) {
            SocketFactory.log.info("begin socket create,{}:{}", connection.getIpAddress(),
                    connection.getPort());
        }
        final Socket socket = new Socket();
        socket.setTrafficClass(0x04 | 0x10);
        socket.setKeepAlive(true);
        socket.setTcpNoDelay(true);
        socket.setSendBufferSize(1024);
        socket.setReceiveBufferSize(1024);
        socket.setSoTimeout(
                connection.getSotimeout() < 100 ? connection.getSotimeout() * 1000 : connection.getSotimeout());
        this.connect(socket, connection, connection.getConnectRetry());
        return socket;

    }

    /**
     * retry mechanism when connect
     *
     * @param socket Socket
     * @param connection SocketConnection
     * @param availableRetryTimes int
     * @throws IOException
     */
    protected void connect(final Socket socket, final SocketConnection connection, int availableRetryTimes)
            throws IOException {
        try {
            socket.connect(new InetSocketAddress(connection.getIpAddress(), connection.getPort()),
                    connection.getConnectTimeout() * (connection.getConnectTimeout() < 100
                            ? connection.getConnectTimeout() * 1000 : connection.getConnectTimeout()));
            if (SocketFactory.log.isInfoEnabled()) {
                SocketFactory.log.info("socket connect is success,{}:{}", connection.getIpAddress(),
                        connection.getPort());
            }
        } catch (final IOException e) {
            if (availableRetryTimes > 0) {
                SocketFactory.log.warn("socket connect is failure,{}:{}, will retry after 1 seconds. {}",
                        connection.getIpAddress(), connection.getPort(), e.getMessage());
                try {
                    Thread.sleep(1000);
                } catch (final InterruptedException e1) {
                    SocketFactory.log.error("", e1);
                }
                this.connect(socket, connection, --availableRetryTimes);
            } else {
                throw e;
            }
        }
    }

    @Override
    public void destroy(final Socket socket) {
        if (SocketFactory.log.isInfoEnabled()) {
            SocketFactory.log.info("begin socket close，{}", socket.getRemoteSocketAddress());
        }
        try {
            socket.close();
        } catch (final IOException e) {
            SocketFactory.log.error(e.getMessage(), e);
        }
        if (SocketFactory.log.isInfoEnabled()) {
            SocketFactory.log.info("socket closed，{}", socket.getRemoteSocketAddress());
        }

    }

    @Override
    public boolean validateObject(final SocketConnection key, final PooledObject<Socket> p) {
        String reason = "socket is ok.";
        boolean validated = p.getObject().isConnected() && !p.getObject().isClosed();
        if (!validated) {
            reason = "socket is not connected or closed.";
        }
        if (key.isUrgentTest()) {
            try {
                p.getObject().sendUrgentData(0xFF);
            } catch (final IOException e) {
                validated = false;
                reason = "failed to send urgent data.";
            }
        }

        if (SocketFactory.log.isDebugEnabled()) {
            SocketFactory.log.debug("validate socket object result：{}，{}", key, reason);
        }
        return validated;
    }

}
